package com.flute.haflute.monitor;

import static java.util.concurrent.TimeUnit.SECONDS;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.flute.haflute.agent.job.JobResourceManager;
import com.flute.haflute.alarm.AlarmService;
import com.flute.haflute.common.ClusterContext;
import com.flute.haflute.exception.JobStateException;
import com.flute.haflute.jobbox.base.StatusReporter.JOB_STATUS;
import com.flute.haflute.tools.PlaceOnFileService;
import com.flute.haflute.tools.SerializeHistroyJob;

public class ClusterStatusMonitor {
	private static Logger logger = LoggerFactory
			.getLogger(ClusterStatusMonitor.class);

	static List<JobMonitorInfo> finishendJoblist = new ArrayList<JobMonitorInfo>();
	static ConcurrentMap<String, JobMonitorInfo> histroyjobMap = new ConcurrentHashMap<String, JobMonitorInfo>();
	static ConcurrentMap<String, List<JobMonitorInfo>> histroyjobMapbyip = new ConcurrentHashMap<String, List<JobMonitorInfo>>();
	static ConcurrentMap<String, Map<String, List<JobMonitorInfo>>> currentmap = new ConcurrentHashMap<String, Map<String, List<JobMonitorInfo>>>();
	static ConcurrentMap<String, Map<String, SystemAbility>> systemmap = new ConcurrentHashMap<String, Map<String, SystemAbility>>();
	static ConcurrentMap<String, StatusMessage> monistorStatusmMap = new ConcurrentHashMap<String, StatusMessage>();
	static ConcurrentMap<String, StatusMessage> monistorStatusmHistoryMap = new ConcurrentHashMap<String, StatusMessage>();
	List<JobMonitorInfo> storageList = new ArrayList<JobMonitorInfo>();

	final ScheduledExecutorService scheduler = Executors
			.newScheduledThreadPool(1);

	final Runnable placeOnFileTask = new Runnable() {
		public void run() {
			Iterator<String> it = monistorStatusmMap.keySet().iterator();
			while (it.hasNext()) {
				String key = it.next();
				StatusMessage statusMessage = monistorStatusmMap.get(key);
				try {
					SimpleDateFormat df = new SimpleDateFormat(
							"yyyy-MM-dd HH:mm:ss");
					long reciiveTime = df.parse(
							statusMessage.reportDate.toString()).getTime();
					long currentTime = System.currentTimeMillis();
					int time = ClusterContext.getMonitorMsgAge();
					String historyKey = key + "^" + reciiveTime;
					if ((currentTime - reciiveTime) / 1000 / 60 > time) {
						if (!monistorStatusmHistoryMap.containsKey(historyKey)) {
							monistorStatusmHistoryMap.put(historyKey,
									monistorStatusmMap.get(key));
							if (statusMessage.topic != null
									&& !statusMessage.topic.trim()
											.equalsIgnoreCase("Jvm")) {
								PlaceOnFileService.getInstance().save(key,
										monistorStatusmMap.get(key));
							}
						}
						monistorStatusmMap.remove(key);
					}
					statusMessage = null;
				} catch (ParseException e) {
					e.printStackTrace();
				}
			}
		}
	};

	/**
	 * triggered when job state changed
	 * 
	 * @param state
	 * @throws JobStateException
	 */
	@SuppressWarnings("unchecked")
	public synchronized void jobStateFinished(JobMonitorInfo jobInfo)
			throws JobStateException {
		logger.info("one Job {} Finished from {}", jobInfo.getJobDescription(),
				jobInfo.getMaster());
		JobResourceManager.getInstance().releaseNodes4Job(jobInfo.getMaster());
		// ------------------------
		// 联系历史的JOB和相关他的监控信息
		Iterator<String> it = monistorStatusmMap.keySet().iterator();
		while (it.hasNext()) {
			String key = it.next();
			if (key.startsWith(jobInfo.getMaster().getIp())
					&& key.indexOf("Jvm") == -1) {
				jobInfo.setStatusMessage(monistorStatusmMap.get(key));
			}
		}
		// ------------------------

		histroyjobMap.put(jobInfo.jobid, jobInfo);
		synchronized (finishendJoblist) {
			finishendJoblist.add(jobInfo);
		}
		SerializeHistroyJob serializeHistroyJob = SerializeHistroyJob
				.getInstance();
		Object obj = serializeHistroyJob.uncoilSerializeObj();
		if (obj instanceof List) {
			List<JobMonitorInfo> list = (List<JobMonitorInfo>) obj;
			for (JobMonitorInfo jinfo : list) {
				if (!finishendJoblist.contains(jinfo))
					finishendJoblist.add(jinfo);
			}
		}

		serializeHistroyJob.saveSerializeObj(finishendJoblist);
		logger.debug(" removeRunningjob4curren -> "
				+ jobInfo.getMaster().getIp() + " jobid = "
				+ jobInfo.getJobid());
		removeRunningjob4curren(jobInfo.getMaster().getIp(), jobInfo);
		histroyjobMapbyip.put(jobInfo.getMaster().getIp(), finishendJoblist);
		AlarmService alarm = new AlarmService();
		alarm.reportFinishendJob(jobInfo);
		alarm = null;
	}

	/**
	 * 定义TEMP MAP 保存KEY VAL 第二次调用的时候用于覆盖同一个MASTER的任务信息 Map<String,
	 * List<JobMonitorInfo> String:IP
	 * 
	 * @param map
	 */
	public synchronized void setJobInfo(Map<String, List<JobMonitorInfo>> map) {
		String key = "";
		Iterator<String> it = map.keySet().iterator();
		while (it.hasNext()) {
			key = it.next();
		}
		// //存储每个任务
		// Iterator<Entry<String, List<JobMonitorInfo>>> listIt =
		// map.entrySet().iterator();
		// while(listIt.hasNext()){
		// Entry<String, List<JobMonitorInfo>> entry = listIt.next();
		// List<JobMonitorInfo> tasklist = entry.getValue();
		// if(tasklist != null){
		// for(JobMonitorInfo jobinfo:tasklist){
		// storageList.add(jobinfo);
		// }
		// }else{
		// isCanPutMap = false;
		// }
		//
		// }
		// if(isCanPutMap)
		currentmap.put(key, map);
		scheduler.scheduleAtFixedRate(placeOnFileTask, 1, 1 * 1, SECONDS);
	}

	public synchronized void setSystemInfo(Map<String, SystemAbility> map) {
		String key = "";
		Iterator<String> it = map.keySet().iterator();
		while (it.hasNext()) {
			key = it.next();
		}
		systemmap.put(key, map);
	}

	/**
	 * 页面调用方法获取正在执行的JOB状态
	 * 
	 * @return
	 */
	public static List<JobMonitorInfo> getJobInfo(String ip) {
		List<JobMonitorInfo> list = new ArrayList<JobMonitorInfo>();
		Iterator<Entry<String, Map<String, List<JobMonitorInfo>>>> it = currentmap
				.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, Map<String, List<JobMonitorInfo>>> e = it.next();
			Map<String, List<JobMonitorInfo>> map = e.getValue();
			list = map.get(ip);
			if (list != null) {
				break;
			}
		}
		return list;
	}

	/**
	 * 页面调用方法获取正在执行的JOB状态
	 * 
	 * @return
	 */
	public static List<JobMonitorInfo> getJobInfo() {
		List<JobMonitorInfo> list = new ArrayList<JobMonitorInfo>();
		Iterator<Entry<String, Map<String, List<JobMonitorInfo>>>> it = currentmap
				.entrySet().iterator();
		while (it.hasNext()) {
			Entry<String, Map<String, List<JobMonitorInfo>>> e = it.next();
			Map<String, List<JobMonitorInfo>> map = e.getValue();
			Iterator<Entry<String, List<JobMonitorInfo>>> itor = map.entrySet()
					.iterator();
			while (itor.hasNext()) {
				Entry<String, List<JobMonitorInfo>> entry = itor.next();
				List<JobMonitorInfo> jlist = entry.getValue();
				if (jlist != null) {
					Iterator<JobMonitorInfo> jit = jlist.iterator();
					while (jit.hasNext()) {
						JobMonitorInfo j = jit.next();
						list.add(j);
					}
				}
			}
		}
		return list;
	}

	/**
	 * 历史JOB
	 * 
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static List<JobMonitorInfo> getHistroyJobInfo(String ip) {

		List<JobMonitorInfo> list = new ArrayList<JobMonitorInfo>();
		SerializeHistroyJob serializeobj = new SerializeHistroyJob(null);
		Object obj = serializeobj.uncoilSerializeObj();
		if (obj instanceof List) {
			list = (List<JobMonitorInfo>) obj;
		}
		serializeobj = null;
		Iterator<JobMonitorInfo> its = list.iterator();
		while (its.hasNext()) {
			JobMonitorInfo jinfo = its.next();
			if (!jinfo.getMaster().getIp().equals(ip)) {
				its.remove();
			}
		}

		if (finishendJoblist != null) {
			Iterator<JobMonitorInfo> it = finishendJoblist.iterator();
			while (it.hasNext()) {
				JobMonitorInfo jinfo = it.next();
				if (jinfo.getMaster().getIp().equals(ip)
						&& !list.contains(jinfo)) {
					list.add(jinfo);
				}
			}
		}
		return list;
	}

	/**
	 * 历史JOB
	 * 
	 * @return
	 */
	@SuppressWarnings("unchecked")
	public static List<JobMonitorInfo> getHistroyJobInfo() {

		List<JobMonitorInfo> list = new ArrayList<JobMonitorInfo>();
		SerializeHistroyJob serializeobj = new SerializeHistroyJob(null);
		Object obj = serializeobj.uncoilSerializeObj();
		if (obj instanceof List) {
			list = (List<JobMonitorInfo>) obj;
		}
		serializeobj = null;

		if (histroyjobMap != null) {
			Iterator<Entry<String, JobMonitorInfo>> it = histroyjobMap
					.entrySet().iterator();
			while (it.hasNext()) {
				Entry<String, JobMonitorInfo> e = it.next();
				JobMonitorInfo histroyjob = e.getValue();
				if (histroyjob.getJobstatus() == JOB_STATUS.FINISHED
						&& !list.contains(histroyjob)) {
					list.add(histroyjob);
				}
			}
		}

		return list;

	}

	public static List<JobMonitorInfo> getAllJobInfo() {
		List<JobMonitorInfo> historylist = new ArrayList<JobMonitorInfo>();
		historylist = getHistroyJobInfo();
		if (getJobInfo() != null) {
			historylist.addAll(getJobInfo());
		}
		return historylist;
	}

	public static SystemAbility getSystemAbility(String ip) {
		Map<String, SystemAbility> map = systemmap.get(ip);
		if (map != null) {
			return map.get(ip);
		} else {
			return null;
		}
	}

	/**
	 * KEY: ip + "^" + status.topic
	 * 
	 * @param map
	 */
	public void receiveStatusMessage(Map<String, StatusMessage> map) {
		Iterator<String> it = map.keySet().iterator();
		while (it.hasNext()) {
			String key = it.next();
			StatusMessage statusMessage = map.get(key);
			// 统一监控消息汇报时间
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			statusMessage.reportDate = sdf.format(new java.util.Date());
			ClusterStatusMonitor.monistorStatusmMap.put(key, map.get(key));
			statusMessage = null;
		}
	}

	public static List<UniversalMonitorMsg> getReceiveStatusMessage(
			String nodeip) {
		Iterator<String> it = monistorStatusmMap.keySet().iterator();
		List<UniversalMonitorMsg> list = new ArrayList<UniversalMonitorMsg>();
		while (it.hasNext()) {
			String key = it.next();
			String moudle = key.split("\\^")[0];
			String ip = moudle.split("-")[0];
			// 如果入参没有指定 就取所有的告警信息，否则只取指定的
			if (nodeip != null) {
				if (nodeip.equals(ip)) {
					StatusMessage statusMsg = monistorStatusmMap.get(key);
					UniversalMonitorMsg univerMsg = new UniversalMonitorMsg();
					univerMsg.setIp(ip);
					univerMsg.setMoudle(moudle);
					univerMsg.setStatusMsg(statusMsg);
					if (ClusterContext.getTopics().containsKey(statusMsg.topic)) {
						univerMsg.order = ClusterContext.getTopics().get(
								statusMsg.topic);
					}
					if (!statusMsg.isFinish) {
						list.add(univerMsg);
					}

				}
			} else {
				StatusMessage statusMsg = monistorStatusmMap.get(key);
				UniversalMonitorMsg univerMsg = new UniversalMonitorMsg();
				univerMsg.setIp(ip);
				univerMsg.setMoudle(moudle);
				univerMsg.setStatusMsg(statusMsg);
				if (ClusterContext.getTopics().containsKey(statusMsg.topic)) {
					univerMsg.order = ClusterContext.getTopics().get(
							statusMsg.topic);
				}
				list.add(univerMsg);
			}

		}
		return list;
	}

	/**
	 * 清除在途的任务记录（在收到结束的任务消息之后）
	 */
	public void removeRunningjob4curren(String ip, JobMonitorInfo jobInfo) {

		Map<String, List<JobMonitorInfo>> map = currentmap.get(ip);
		if (map == null) {
			return;
		}
		List<JobMonitorInfo> list = map.get(ip);
		if (list == null) {
			return;
		}
		Iterator<JobMonitorInfo> it = list.iterator();
		while (it.hasNext()) {
			JobMonitorInfo jobMonitorInfo = it.next();
			if (jobMonitorInfo.equals(jobInfo)) {
				logger.info("remove job id = " + jobMonitorInfo.getJobid());
				it.remove();
			}
		}
		HashMap<String, List<JobMonitorInfo>> temp = new HashMap<String, List<JobMonitorInfo>>();
		temp.put(ip, list);
		currentmap.put(ip, temp);
	}

	/**
	 * 查询当天记录 入参填NULL 否则填入日期，格式如下
	 * 
	 * @param day
	 *            - > "yyyyMMdd"
	 * @return
	 */
	public static List<UniversalMonitorMsg> getHistoryMonitorInfo(String day) {
		return PlaceOnFileService.getInstance().readText(day);
	}

}
